package dnet

import (
	"dgo/goutils/utils"
	"fmt"
	"net"
	"sync"
	"sync/atomic"
	"time"
)

type EventBufferEx func(client *ASyncTcpClientEx, buffer []byte)
type EventTcpClientEx func(client *ASyncTcpClientEx)
type EventTcpClientResultEx func(client *ASyncTcpClientEx) interface{}

/*
**

	发送使用workers版本
	 OnConnected, OnDisconnected 只在 接收纤程中触发
	推荐使用ASyncTcpClientEx
*/
type ASyncTcpClientEx struct {
	sendBytes   int64 // 注意64位对齐  32位下
	connectflag int32
	startflag   int32

	object_id uint32

	start_t time.Time
	stop_t  time.Time

	IsTerminated bool

	RecvInterval time.Duration

	Conn              net.Conn
	RemoteAddress     string
	RecvBlockSize     uint16
	OnConnected       EventTcpClientEx
	OnDisconnected    EventTcpClientEx // 只在recv纤程中触发,其他地方不进行触发
	OnConnectFail     EventTcpClientEx
	OnRecvBuffer      EventBufferEx
	OnErrEvent        ErrEvent
	LastConnectErr    error
	UseQueueSend      bool // 使用队列进行发送
	LastDisMsg        string
	closeWg           *sync.WaitGroup
	lastSendTime      time.Time
	connectTime       time.Time
	disTime           time.Time     // 断线时间
	reconnectInterval time.Duration // 重连间隔
	LastRecvTime      time.Time     // 最后一次接收数据的时间
	BeforeConnectCb   EventTcpClientResultEx
	DailCb            func(network, addr string) (c net.Conn, err error)
}

func NewASyncTcpClientEx() *ASyncTcpClientEx {
	client := &ASyncTcpClientEx{
		object_id:         utils.NewObjectDNA(),
		IsTerminated:      false,
		startflag:         0,
		reconnectInterval: time.Second * 5,
		RecvBlockSize:     RECV_BLOCK_SIZE,
		UseQueueSend:      true,
		closeWg:           new(sync.WaitGroup),
	}
	return client
}

func (this *ASyncTcpClientEx) GetObjectID() uint32 {
	return this.object_id
}

func (this *ASyncTcpClientEx) SendBytes() int64 {
	return this.sendBytes
}

func (this *ASyncTcpClientEx) SetAddress(addr string) {
	this.RemoteAddress = addr
}

func (this *ASyncTcpClientEx) IsActive() bool {
	return this.startflag == 1
}

/*
start 和Stop 最好在同一个纤程中运行. 不要进行并发
*/
func (this *ASyncTcpClientEx) Start() {
	if atomic.CompareAndSwapInt32(&this.startflag, 0, 1) {
		this.start_t = time.Now()
		this.IsTerminated = false
		utils.GoFunc2(this.handRecv)
		time.Sleep(time.Millisecond) // 等待开启
	} else {
		utils.Log.Warn("连接已经打开！请勿重复启动")
	}
}

func (this *ASyncTcpClientEx) Stop() {
	this.IsTerminated = true
	this.innerCheckClose()
}

func (this *ASyncTcpClientEx) GetConnectedTime() time.Time {
	return this.connectTime
}

func (this *ASyncTcpClientEx) GetStartTime() time.Time {
	return this.start_t
}

func (this *ASyncTcpClientEx) GetWorkTimeInfo() string {
	if this.startflag == 1 {
		if !this.stop_t.IsZero() {
			return fmt.Sprintf("->[%s] [%s]->[NOW]", utils.DateTimeStringForShortShow(this.stop_t), utils.DateTimeStringForShortShow(this.start_t))
		} else {
			return fmt.Sprintf("[%s]->[NOW]", utils.DateTimeStringForShortShow(this.start_t))
		}
	} else {
		if this.start_t.IsZero() {
			return ""
		} else {
			return fmt.Sprintf(" %s->[%s]", utils.DateTimeStringForShortShow(this.start_t), utils.DateTimeStringForShortShow(this.stop_t))
		}
	}

}

func (this *ASyncTcpClientEx) GetStopTime() time.Time {
	return this.stop_t
}

func (this *ASyncTcpClientEx) RequestDisconnect(reason string) {
	conn := this.Conn
	if conn == nil {
		return
	}
	if atomic.CompareAndSwapInt32(&this.connectflag, 1, 0) {
		this.LastDisMsg = fmt.Sprintf("[%s]%s", utils.DateTimeStringForShortShow(time.Now()), reason)
		conn.Close()
	}
}

func (this *ASyncTcpClientEx) SendBufferDirect(buffer []byte) {
	defer utils.PanicHandlerWithPrint()
	conn := this.Conn
	if conn == nil {
		if this.OnErrEvent != nil {
			this.OnErrEvent(this, fmt.Errorf("连接已经被关闭"), ERR_TYPE_IO_WRITE)
		}
		return
	}

	n, err := conn.Write(buffer)
	if err != nil {
		if this.OnErrEvent != nil {
			this.OnErrEvent(this, err, ERR_TYPE_IO_WRITE)
		}
		return
	} else {
		atomic.AddInt64(&this.sendBytes, int64(n))
		this.lastSendTime = time.Now()
	}

}

func (this *ASyncTcpClientEx) PostSendBuffer(buffer []byte) int {
	if this.startflag == 1 {
		if this.UseQueueSend {
			utils.DefaultWorkers().PostTaskFunc(this.object_id, func() {
				if this.startflag == 0 {
					if this.OnErrEvent != nil {
						this.OnErrEvent(this, fmt.Errorf("连接已经被关闭"), ERR_TYPE_IO_WRITE)
					}
					return
				}

				conn := this.Conn
				if conn == nil {
					if this.OnErrEvent != nil {
						this.OnErrEvent(this, fmt.Errorf("连接已经被关闭"), ERR_TYPE_IO_WRITE)
					}
					return
				}

				this.closeWg.Add(1)
				defer this.closeWg.Done()
				n, err := conn.Write(buffer)
				if err != nil {
					if this.OnErrEvent != nil {
						this.OnErrEvent(this, err, ERR_TYPE_IO_WRITE)
					}
					this.RequestDisconnect(err.Error())
				} else {
					atomic.AddInt64(&this.sendBytes, int64(n))
					this.lastSendTime = time.Now()
				}

			})
			return len(buffer)
		} else {
			n, err := this.Conn.Write(buffer)
			if err != nil {
				if this.OnErrEvent != nil {
					this.OnErrEvent(this, err, ERR_TYPE_IO_WRITE)
				}
				this.RequestDisconnect(err.Error())
			}
			atomic.AddInt64(&this.sendBytes, int64(n))
			this.lastSendTime = time.Now()

			return n
		}

	} else {
		utils.Log.Debugf("连接(%s)已经断开，不投递到发送队列！", this.RemoteAddress)
		return 0
	}
}

func (this *ASyncTcpClientEx) GetRemoteAddress() string {
	conn := this.Conn
	if conn != nil {
		return conn.RemoteAddr().String()
	} else {
		return this.RemoteAddress
	}
}

func (this *ASyncTcpClientEx) checkConnect() net.Conn {
	conn := this.Conn
	if conn != nil {
		return conn
	}

	if this.IsTerminated {
		return nil
	}

	//  重连时间间隔检测
	if !this.disTime.IsZero() && time.Since(this.disTime) < this.reconnectInterval {
		return nil
	}

	if this.BeforeConnectCb != nil {
		r := this.BeforeConnectCb(this)
		if r.(bool) {

		} else {
			return nil
		}
	}
	var err error
	//if this.DailCb != nil {
	//	conn, err = this.DailCb("tcp", this.RemoteAddress)
	//} else {
	//	conn, err = net.Dial("tcp", this.RemoteAddress)
	//}

	if this.DailCb != nil {
		conn, err = this.DailCb("tcp", this.RemoteAddress)
	}

	if err == nil && conn == nil {
		conn, err = net.Dial("tcp", this.RemoteAddress)
	}

	if err != nil {
		if this.OnErrEvent != nil {
			this.OnErrEvent(this, err, ERR_TYPE_CONNECT)
		}
		this.LastConnectErr = err
		if this.OnConnectFail != nil {
			this.OnConnectFail(this)
		}
		return nil
	}
	this.Conn = conn
	this.connectflag = 1
	if this.OnConnected != nil {
		this.OnConnected(this)
	}
	this.connectTime = time.Now()
	this.LastRecvTime = utils.ZeroTime

	return conn
}

func (this *ASyncTcpClientEx) innerCheckClose() {
	if atomic.CompareAndSwapInt32(&this.startflag, 1, 0) {
		this.stop_t = time.Now()
		this.IsTerminated = true
		// utils.Log.Debugf("[%d][%s]主动关闭连接", this.object_id, this.RemoteAddress)

		if utils.DEBUG_MODE == 1 {
			//callstackstr := utils.GetCallStack(1)
			this.RequestDisconnect(fmt.Sprintf("连接请求关闭, sn:%d", utils.NewOperaSN()))
		} else {
			this.RequestDisconnect("连接请求关闭")
		}

		this.closeWg.Wait()
	}
}

// / 如果超时没有收到数据(返回true
func (this *ASyncTcpClientEx) CheckSendTimeOut(isec int) bool {
	if this.connectflag == 0 {
		return false
	}

	t1 := this.lastSendTime
	if t1.IsZero() || this.connectTime.After(t1) { // 使用连接时间
		t1 = this.connectTime
	}

	t := time.Since(t1)
	tick := int(t.Seconds())
	if tick > isec {
		return true
	} else {
		return false
	}
}

// / 如果超时没有收到数据(返回true
func (this *ASyncTcpClientEx) CheckRecvTimeOut(isec int) bool {
	if this.connectflag == 0 {
		return false
	}

	t1 := this.LastRecvTime
	if t1.IsZero() || this.connectTime.After(t1) { // 没有接收到数据就用连接时间
		t1 = this.connectTime
	}

	t := time.Since(t1)
	tick := int(t.Seconds())
	if tick > isec {
		return true
	} else {
		return false
	}
}

func (this *ASyncTcpClientEx) CheckRecvTimeOutReason(isec int) string {
	if this.connectflag == 0 {
		return ""
	}

	t1 := this.LastRecvTime
	if t1.IsZero() || this.connectTime.After(t1) { // 没有接收到数据就用连接时间
		t1 = this.connectTime
	}

	t := time.Since(t1)
	tick := int(t.Seconds())
	if tick > isec {
		return fmt.Sprintf("超时[%d s](%s)未收到数据", tick, utils.DateTimeStringForShortShow(t1))
	} else {
		return ""
	}
}

// / 如果返回true, 表示断开了一次连接
func (this *ASyncTcpClientEx) CheckBreakConnectIfTimeOut(isec int) bool {
	recvTimeOut := this.CheckRecvTimeOut(isec)
	sendTimeOut := this.CheckSendTimeOut(isec)
	if recvTimeOut && sendTimeOut {
		this.RequestDisconnect("超时没有数据请求断线")
		return true
	} else {
		return false
	}
}

// / 如果返回true, 表示断开了一次连接
func (this *ASyncTcpClientEx) CheckBreakConnectIfRecvTimeOut(isec int) bool {
	if this.CheckRecvTimeOut(isec) {
		this.RequestDisconnect("接收数据超时...")
		return true
	} else {
		return false
	}
}

func (this *ASyncTcpClientEx) Connected() bool {
	return this.connectflag == 1
}

func (this *ASyncTcpClientEx) handRecv() {
	this.closeWg.Add(1)
	defer this.closeWg.Done()

recvfor:
	for {
		conn := this.checkConnect()
		if conn == nil {
			if this.IsTerminated {
				break recvfor
			}
			time.Sleep(time.Second)
			continue
		}

		buffer := make([]byte, this.RecvBlockSize)
		n, err := conn.Read(buffer)
		if err != nil {
			if this.OnErrEvent != nil {
				this.OnErrEvent(this, err, ERR_TYPE_IO_RECV)
			}

			if this.OnDisconnected != nil {
				this.OnDisconnected(this)
			}

			this.RequestDisconnect(fmt.Sprintf("接收数据出现异常:%s", err.Error()))
			this.Conn = nil

			if this.IsTerminated {
				break recvfor
			} else {
				time.Sleep(time.Second) // 休息一秒
			}
		} else {
			if this.IsTerminated {
				this.RequestDisconnect("任务终止,请求断线")
				continue
			}
			this.LastRecvTime = time.Now()
			if this.OnRecvBuffer != nil {
				this.OnRecvBuffer(this, buffer[:n])
			}
		}
		if this.RecvInterval > 0 {
			time.Sleep(this.RecvInterval)
		}
		//time.Sleep(time.Second * 10)
	}

	// utils.Log.Debugf("[%d][%s]循环接收任务工作退出！", this.object_id, this.RemoteAddress)
}
